package com.lucien.framework.rabbitmq.config;

import com.lucien.framework.exception.GlobalException;
import com.lucien.framework.rabbitmq.constants.RabbitCons;
import com.lucien.framework.web.chat.ChatService;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Author: Lucien
 * @Description:
 * @Date: Created in 2021/5/3
 */
@Configuration
public class RabbitMqConfig {

    private static final Logger log = LoggerFactory.getLogger(RabbitMqConfig.class);

    @Autowired
    private ChatService chatService;

    /**
     * 绑定键
     */
    private static final String msgTopicKey = "topic.chatroom";

    /**
     * 队列
     */
    private static final String msgTopicQueue = "topicChatQueue";

    /**
     * 交换机
     */
    private static final String msgTopicExchange = "topicChatroomExchange";

    @Bean
    public Queue chatroomQueue() {
        return new Queue(RabbitCons.TOPIC_QUEUE, false, true, true);
    }

    @Bean
    TopicExchange chatroomExchange() {
        return new TopicExchange(RabbitCons.TOPIC_EXCHANGE, false, true);
    }

    @Bean
    Binding binding() {
        return BindingBuilder.bind(chatroomQueue()).to(chatroomExchange()).with(RabbitCons.TOPIC_ROUTING_KEY);
    }

    /**
     * 创建连接工厂，设置工厂信息并设置消息回调
     *
     * @return
     */
    @Bean
    public ConnectionFactory connectionFactory() {
        //连接rabbitmq的信息
        CachingConnectionFactory factory = new CachingConnectionFactory("127.0.0.1", 5672);
        factory.setUsername("guest");
        factory.setPassword("guest");

        // 发送消息回调,必须要设置
        //开启发送消息回调
        factory.setPublisherConfirms(true);
        factory.setPublisherReturns(true);

        return factory;
    }

    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);

        //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);

        /**
         * correlationData  相关数据
         * ack              确认情况
         * cause            原因
         */
        rabbitTemplate.setConfirmCallback
                ((correlationData, ack, cause) ->
                        log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause));

        /**
         * message      消息
         * replyCode    回应码
         * replyText    回应信息
         * exchange     交换机
         * routingKey   路由键
         */
        rabbitTemplate.setReturnCallback
                ((message, replyCode, replyText, exchange, routingKey) ->
                        log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message));


        log.info("===== rabbitmqConfig注入 =====");
        return rabbitTemplate;
    }


//    @Bean
//    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
//        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
//        factory.setConnectionFactory(connectionFactory);
//        factory.setMessageConverter(new Jackson2JsonMessageConverter());
//        return factory;
//    }

    /**
     * 接受消息的监听，这个监听会接受消息队列chatroomQueue的消息
     * 针对消费者配置
     *
     * @return
     */
//    @Bean
//    public SimpleMessageListenerContainer messageContainer() {
//        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
//        //监听的队列信息
//        container.setQueues(chatroomQueue());
//        container.setExposeListenerChannel(true);
//        container.setMaxConcurrentConsumers(1);
//        container.setConcurrentConsumers(1);
//        //设置确认模式手工确认
//        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
//
//        //消费了消息后的处理
//        container.setMessageListener(new ChannelAwareMessageListener() {
//            @Override
//            public void onMessage(Message message, Channel channel) throws Exception {
//                byte[] body = message.getBody();
//                String msg = new String(body);
//                System.out.println("rabbitmq收到消息 : " + msg);
//                Boolean sendToWebsocket = chatService.sendMsg(msg);
//
//                if (sendToWebsocket) {
//                    System.out.println("消息处理成功！ 已经推送到websocket！");
//                    //确认消息成功消费
//                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
//                }
//            }
//        });
//
//        return container;
//    }

}
